package com.john.study.es.document;

import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.john.study.es.config.matcher.PatternMatcher;
import com.john.study.es.fetcher.FetchContext;
import com.john.study.es.fetcher.WebFetcher;
import com.john.study.es.module.ChinaUnicomAllPageVo;
import org.apache.commons.lang3.StringUtils;
import org.elasticsearch.action.index.IndexResponse;
import org.elasticsearch.client.transport.TransportClient;
import org.elasticsearch.common.xcontent.XContentType;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Random;

/**
 * @author jiangguangtao 2017/7/29.
 */
public class DocumentRunner implements Runnable {
    private static final Logger LOGGER = LoggerFactory.getLogger(DocumentRunner.class);

    private static final Random RANDOM = new Random();
    private TransportClient client;
    private String storeIndex;
    private String storeType;
    private String url;
    private PatternMatcher urlScopeMatcher;// 查询范围正则

    private boolean useDelay = true; // 每一个页面处理后在处理下一个页面时是否有延迟

    public DocumentRunner(TransportClient client, String storeIndex, String storeType, String url, PatternMatcher urlScopeMatcher) {
        this.client = client;
        this.storeIndex = storeIndex;
        this.storeType = storeType;
        this.url = url;
        this.urlScopeMatcher = urlScopeMatcher;
    }

    public boolean isUseDelay() {
        return useDelay;
    }

    public void setUseDelay(boolean useDelay) {
        this.useDelay = useDelay;
    }

    private int getNextDelay() {
        // [500-5500）之间的随机数
        return RANDOM.nextInt(3000) + 500;
    }

    @Override
    public void run() {

        FetchContext fc = new FetchContext(urlScopeMatcher);
        fc.setUrl(this.url);
        fc.setFetchUrls(true);

        boolean flag = fetchPageVo(fc);
        String nextUrl = fc.getNextUrl();

        while (StringUtils.isNotBlank(nextUrl)) {
            if (isUseDelay()) {
                try {
                    int nextDelay = getNextDelay();
                    LOGGER.debug("下次请求前延迟 {}ms", nextDelay);
                    Thread.sleep(nextDelay);
                } catch (InterruptedException e) {
                    LOGGER.error("中断异常", e);
                }
            }
            fc.setUrl(nextUrl);
            flag = fetchPageVo(fc);
            nextUrl = fc.getNextUrl();
        }
    }

    /**
     * @param fc
     * @return true 成功处理，FALSE处理失败
     * @throws JsonProcessingException
     */
    private boolean fetchPageVo(FetchContext fc) {
        WebFetcher fetcher = new WebFetcher();

        ChinaUnicomAllPageVo pageVo = fetcher.fetcher(fc);
        if (null == pageVo) {
            return false;
        }
        try {
            ObjectMapper objectMapper = new ObjectMapper();
            String json = objectMapper.writeValueAsString(pageVo);
            LOGGER.trace("待创建的对象是：\n{}", json);
            LOGGER.debug("创建索引：{} {} {}", pageVo.getUrlKey(), pageVo.getUrl(), pageVo.getTitle());

            IndexResponse indexResponse = client.prepareIndex(this.storeIndex, this.storeType, pageVo.getUrlKey())
                    .setSource(json, XContentType.JSON)
                    .get();
            LOGGER.debug("提交文档返回值：{}", indexResponse);
            return true;
        } catch (Exception e) {
            LOGGER.error("处理url失败", e);
            return false;
        }
    }
}
